package edu.utexas.mapreduce;

import java.net.MalformedURLException;
import java.net.URL;
import java.util.Map;
import java.util.Set;
import java.util.StringTokenizer;
import java.util.TreeMap;
import java.util.Map.Entry;

import edu.utexas.ipc.Connection;
import edu.utexas.ipc.Message;
import edu.utexas.ipc.MessageListener;
import edu.utexas.ipc.Mutex;

public class JobStateGlobal implements JobState, MessageListener {

	JobState jobState;
	Mutex mutex;
	Connection connection;
	
	JobStateGlobal(JobState jobState, Mutex mutex, Connection connection) {
		this.jobState = jobState;
		this.mutex = mutex;
		this.connection = connection;
		
		connection.registerListener(this);
	}
	
	public synchronized void commitMapResults(String owner, Map<Integer, URL> results) {
		mutex.acquire();
		jobState.commitMapResults(owner, results);
		
		// serialize map results
		StringBuilder sb = new StringBuilder(); 
		for(Entry<Integer, URL> e : results.entrySet()) {
			if (sb.length() != 0) {
				sb.append("|");
			}
			
			sb.append(e.getKey());
			sb.append("=>");
			sb.append(e.getValue().toString());
		}
		connection.send("complete_map " + owner + " " + sb.toString());
		mutex.release();
	}

	public synchronized void completeReduceTask(String owner, ReduceTask task) {
		mutex.acquire();
		jobState.completeReduceTask(owner, task);
		connection.send("complete_reduce " + task.getID() + " " + owner);
		mutex.release();
	}

	public synchronized int getJobID() {
		// This is immutable data (TODO)
		return jobState.getJobID();
	}


	public synchronized boolean mapTasksComplete(Set<String> activeProcessSet) {
		mutex.acquire();
		boolean complete = jobState.mapTasksComplete(activeProcessSet);
		mutex.release();
		return complete;
	}

	public synchronized boolean reduceTasksComplete(Set<String> activeProcessSet) {
		mutex.acquire();
		boolean complete = jobState.reduceTasksComplete(activeProcessSet);
		mutex.release();
		return complete;
	}

	public synchronized MapTask reserveIdleMapTask(String owner) {
		mutex.acquire();
		MapTask task = jobState.reserveIdleMapTask(owner);
		
		// send result to other processes only if we reserved a task
		if (task != null) {
			connection.send("reserve_map " + task.getID() + " " + owner);
		}
		mutex.release();
		return task;
	}

	public synchronized ReduceTask reserveIdleReduceTask(String owner) {
		mutex.acquire();
		ReduceTask task = jobState.reserveIdleReduceTask(owner);
		
		// send result to other processes only if we reserved a task
		if (task != null) {
			connection.send("reserve_reduce " + task.getID() + " " + owner);
		}
		mutex.release();
		return task;
	}

	public synchronized String getOutputDir() {
		// This is immutable data.  move somewhere else? (TODO)
		return jobState.getOutputDir();
	}

	public synchronized void recoverFromProcessFailure(String failedProcess) {
		mutex.acquire();
		jobState.recoverFromProcessFailure(failedProcess);
		connection.send("recover_failure " + failedProcess);
		mutex.release();
	}

	public void connectionLost(int processID) {
		// no-op
	}

	public void messageReceived(Message msg) {
		try {
			// TODO apply updates to the task table from other processes
			//      If possible check that the source ID is the same process
			//      that owns the mutex.  Or at least check that we are not
			//      the owner of the mutex.

			// System.out.println("state_msg: " + msg);

			StringTokenizer tk = new StringTokenizer(msg.appData);

			final String cmd = tk.nextToken();

			//
			if (cmd.equals("reserve_map")) {
				final int taskID = Integer.parseInt(tk.nextToken());
				final String owner = tk.nextToken();
				jobState.markMapTaskInProgress(taskID, owner);
				return;
			}

			if (cmd.equals("complete_map")) {
				final String owner = tk.nextToken();

				Map<Integer, URL> results = new TreeMap<Integer, URL>();
				while(tk.hasMoreTokens()) {
					final String entry = tk.nextToken("|").trim();
					final int key = Integer.parseInt(entry.split("=>")[0]);
					final URL value = new URL(entry.split("=>")[1]);
					results.put(key, value);
				}

				jobState.commitMapResults(owner, results);
				return;
			}

			if (cmd.equals("release_reduce")) {
				final int taskID = Integer.parseInt(tk.nextToken());
				final String owner = tk.nextToken();
				jobState.markReduceTaskIdle(taskID, owner);
				return;
			}
			
			if (cmd.equals("reserve_reduce")) {
				final int taskID = Integer.parseInt(tk.nextToken());
				final String owner = tk.nextToken();
				jobState.markReduceTaskInProgress(taskID, owner);
				return;
			}
			
			if (cmd.equals("complete_reduce")) {
				final int taskID = Integer.parseInt(tk.nextToken());
				final String owner = tk.nextToken();
				jobState.markReduceTaskComplete(taskID, owner);
				return;
			}
			
			if (cmd.equals("recover_failure")) {
				final String failedProcess = tk.nextToken();
				jobState.recoverFromProcessFailure(failedProcess);
				return;
			}
			
			if (cmd.equals("notification_sent")) {
				jobState.markNotificationSent();
				return;
			}
		}
		catch(MalformedURLException e) {
			e.printStackTrace();
		}
		
	}

	public void close() {
		mutex.close();
		connection.close();
	}
	
	public String toString() {
		return jobState.toString();
	}

	public void markMapTaskInProgress(int id, String owner) {
		mutex.acquire();
		jobState.markMapTaskInProgress(id, owner);
		connection.send("reserve_map " + id + " " + owner);
		mutex.release();
	}

	public void markReduceTaskIdle(int id, String owner) {
		mutex.acquire();
		jobState.markReduceTaskIdle(id, owner);
		connection.send("release_reduce " + id + " " + owner);
		mutex.release();
	}
	
	public void markReduceTaskInProgress(int id, String owner) {
		mutex.acquire();
		jobState.markReduceTaskInProgress(id, owner);
		connection.send("reserve_reduce " + id + " " + owner);
		mutex.release();
	}

	public void markReduceTaskComplete(int id, String owner) {
		mutex.acquire();
		jobState.markReduceTaskComplete(id, owner);
		connection.send("complete_reduce " + id + " " + owner);
		mutex.release();
	}

	public void markNotificationSent() {
		mutex.acquire();
		connection.send("notification_sent");
		mutex.release();
	}

	public void sendNotification() {
		mutex.acquire();
		jobState.sendNotification();
		connection.send("notification_sent");
		mutex.release();
	}


}
